use crate::rtmp::chunk::define::CHUNK_SIZE;
use crate::rtmp::chunk::packetizer::ChunkPacketizer;
use crate::rtmp::chunk::unpacketizer::{ChunkUnpacketizer, UnpackResult};
use crate::rtmp::handshake;
use crate::rtmp::handshake::define::ClientHandshakeState;
use crate::rtmp::handshake::handshake_client::SimpleHandshakeClient;
use crate::rtmp::messages::define::RtmpMessageData;
use crate::rtmp::messages::parser::MessageParser;
use crate::rtmp::net_connection::writer::{ConnectProperties, NetConnection};
use crate::rtmp::net_stream::writer::NetStreamWriter;
use crate::rtmp::protocol_control_messages::writer::ProtocolControlMessagesWriter;
use crate::rtmp::session::common::Common;
use crate::rtmp::session::define;
use crate::rtmp::session::define::SessionType;
use crate::rtmp::session::errors::{SessionError, SessionErrorValue};
use crate::rtmp::user_control_messages::writer::EventMessagesWriter;
use crate::rtmp::utils::RtmpUrlParser;
use bytesio::bytes_writer::AsyncBytesWriter;
use bytesio::bytesio::{TNetIO, TcpIO};
use indexmap::IndexMap;
use std::sync::Arc;
use std::time::Duration;
use streamhub::define::StreamHubEventSender;
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use xflv::amf0::Amf0ValueType;

// 客户端绘画状态枚举
#[allow(dead_code)]
enum ClientSessionState {
    // 握手状态
    Handshake,
    // 连接状态
    Connect,
    // 创建流状态
    CreateStream,
    // 播放状态
    Play,
    // 发布内容状态
    PublishingContent,
    // 开始发布状态
    StartPublish,
    // 等待状态改变
    WaitStateChange,
}

#[allow(dead_code)]
enum ClientSessionPlayState {
    Handshake,
    Connect,
    CreateStream,
    Play,
}

// 客户端会话播放状态枚举
#[allow(dead_code)]
enum ClientSessionPublishState {
    // 握手状态
    Handshake,
    // 连接状态
    Connect,
    // 创建流状态
    CreateStream,
    // 发布内容状态
    PublishingContent,
}
#[allow(dead_code)]
#[derive(Clone, Debug, PartialEq)]
pub enum ClientSessionType {
    // 拉流类型
    Pull,
    // 推流类型
    Push,
}

// 客户端会话结构体
pub struct ClientSession {
    // 网络I/O
    io: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>,
    // 超时时间
    timeout: Option<Duration>,
    // 公共模块
    common: Common,
    // 握手客户端
    handshaker: SimpleHandshakeClient,
    // 块解包器
    unpacketizer: ChunkUnpacketizer,
    // 原始域名（带端口）
    raw_domain_name: String,
    // 应用名称
    app_name: String,
    // 原始流名称（带参数）
    raw_stream_name: String,
    // 流名称
    stream_name: String,
    // 当前状态
    state: ClientSessionState,
    // 客户端类型
    client_type: ClientSessionType,
    // 子应用名称
    sub_app_name: Option<String>,
    // 子流名称
    sub_stream_name: Option<String>,
    // GOP数量
    gop_num: usize,
}

impl ClientSession {
    pub fn new(
        stream: TcpStream,
        client_type: ClientSessionType,
        raw_domain_name: String,
        app_name: String,
        raw_stream_name: String,
        event_producer: StreamHubEventSender,
        gop_num: usize,
    ) -> Self {
        // 获取服务器地址
        let remote_addr = if let Ok(addr) = stream.peer_addr() {
            log::info!("server session: {}", addr.to_string());
            Some(addr)
        } else {
            None
        };

        // 创建 IO 实例
        let tcp_io: Box<dyn TNetIO + Send + Sync> = Box::new(TcpIO::new(stream));
        let net_io = Arc::new(Mutex::new(tcp_io));

        // 根据客户端类型决定是否创建块打包器
        let packetizer = if client_type == ClientSessionType::Push {
            Some(ChunkPacketizer::new(Arc::clone(&net_io)))
        } else {
            None
        };

        // 创建会话通用功能模块实例
        let common = Common::new(packetizer, event_producer, SessionType::Client, remote_addr);
        // 解析流名称，去除参数部分
        let (stream_name, _) = RtmpUrlParser::parse_stream_name_with_query(&raw_stream_name);

        Self {
            io: Arc::clone(&net_io),
            timeout: None,
            common,
            handshaker: SimpleHandshakeClient::new(Arc::clone(&net_io)),
            unpacketizer: ChunkUnpacketizer::new(),
            raw_domain_name,
            app_name,
            raw_stream_name,
            stream_name,
            state: ClientSessionState::Handshake,
            client_type,
            sub_app_name: None,
            sub_stream_name: None,
            gop_num,
        }
    }

    // 设置会话操作的超时时间
    pub fn set_timeout(&mut self, timeout: Duration) {
        self.timeout = Some(timeout)
    }

    // 异步运行客户端会话，处理会话的主要逻辑
    pub async fn run(&mut self) -> Result<(), SessionError> {
        loop {
            match self.state {
                // 当前处于握手状态
                ClientSessionState::Handshake => {
                    log::info!("[C -> S] handshake...");
                    // 执行握手操作，操作完成后继续循环
                    self.handshake().await?;
                    continue;
                }
                // 当前处于连接状态
                ClientSessionState::Connect => {
                    log::info!("[C -> S] connect...");
                    // 发送连接请求
                    self.send_connect(&(define::TRANSACTION_ID_CONNECT as f64))
                        .await?;
                    // 状态切换到等待状态改变
                    self.state = ClientSessionState::WaitStateChange;
                }
                // 当前处于创建流状态
                ClientSessionState::CreateStream => {
                    log::info!("[C -> S] CreateStream...");
                    // 发送创建流请求
                    self.send_create_stream(&(define::TRANSACTION_ID_CREATE_STREAM as f64))
                        .await?;
                    // 状态切换到等待状态改变
                    self.state = ClientSessionState::WaitStateChange;
                }
                // 当前处于播放状态
                ClientSessionState::Play => {
                    log::info!("[C -> S] Play...");
                    // 发送播放请求
                    self.send_play(&0.0, &self.raw_stream_name.clone(), &0.0, &0.0, &false)
                        .await?;
                    // 状态切换到等待状态改变
                    self.state = ClientSessionState::WaitStateChange;
                }
                // 当前处于发布内容状态
                ClientSessionState::PublishingContent => {
                    log::info!("[C -> S] PublishingContent...");
                    // 发送发布请求
                    self.send_publish(&3.0, &self.raw_stream_name.clone(), &"live".to_string())
                        .await?;
                    // 状态切换到等待状态改变
                    self.state = ClientSessionState::WaitStateChange;
                }
                // 当前处于开始发布状态
                ClientSessionState::StartPublish => {
                    log::info!("[C -> S] StartPublish...");
                    // 发送通道数据
                    self.common.send_channel_data().await?;
                }
                // 当前处于等待状态改变状态
                ClientSessionState::WaitStateChange => {}
            }

            // 根据是否设置了超时时间，选择不同的读取方式
            let data = match self.timeout {
                None => self.io.lock().await.read().await?,
                Some(t) => self.io.lock().await.read_timeout(t).await?,
            };
            // 将读取到的数据添加到块解包器中
            self.unpacketizer.extend_data(&data[..]);

            // 循环解包数据
            loop {
                match self.unpacketizer.read_chunks() {
                    Ok(rv) => {
                        if let UnpackResult::Chunks(chunks) = rv {
                            for chunk_info in chunks.iter() {
                                if let Some(mut msg) =
                                    MessageParser::new(chunk_info.clone()).parse()?
                                {
                                    let timestamp = chunk_info.message_header.timestamp;
                                    self.process_messages(&mut msg, &timestamp).await?;
                                }
                            }
                        }
                    }
                    Err(err) => {
                        log::trace!("read trunks error: {}", err);
                        break;
                    }
                }
            }
        }
    }

    // 异步执行握手操作
    async fn handshake(&mut self) -> Result<(), SessionError> {
        loop {
            self.handshaker.handshake().await?;
            if self.handshaker.state == ClientHandshakeState::Finish {
                log::info!("handshake finish");
                break;
            }

            let mut bytes_len = 0;
            while bytes_len < handshake::define::RTMP_HANDSHAKE_SIZE * 2 {
                let data = self.io.lock().await.read().await?;
                bytes_len += data.len();
                self.handshaker.extend_data(&data[..]);
            }
        }

        self.state = ClientSessionState::Connect;

        Ok(())
    }

    // 异步处理接收到的 RTMP 消息
    pub async fn process_messages(
        &mut self,
        msg: &mut RtmpMessageData,
        timestamp: &u32,
    ) -> Result<(), SessionError> {
        match msg {
            // 处理 AMF0 命令消息
            RtmpMessageData::Amf0Command {
                command_name,
                transaction_id,
                command_object,
                others,
            } => {
                log::info!("[C <- S] on_amf0_command_message...");
                // 调用处理 AMF0 命令消息的方法
                self.on_amf0_command_message(command_name, transaction_id, command_object, others)
                    .await?
            }
            // 处理设置对等带宽消息
            RtmpMessageData::SetPeerBandwidth { .. } => {
                log::info!("[C <- S] on_set_peer_bandwidth...");
                // 调用处理设置对等带宽消息的方法
                self.on_set_peer_bandwidth().await?
            }
            // 处理窗口确认大小消息
            RtmpMessageData::WindowAcknowledgementSize { .. } => {
                log::info!("[C <- S] on_windows_acknowledgement_size...");
            }
            // 处理设置块大小消息
            RtmpMessageData::SetChunkSize { chunk_size } => {
                log::info!("[C <- S] on_set_chunk_size...");
                // 调用处理设置块大小消息的方法
                self.on_set_chunk_size(chunk_size)?;
            }
            // 处理流开始消息
            RtmpMessageData::StreamBegin { stream_id } => {
                log::info!("[C <- S] on_stream_begin...");
                // 调用处理流开始消息的方法
                self.on_stream_begin(stream_id)?;
            }
            // 处理流已录制消息
            RtmpMessageData::StreamIsRecorded { stream_id } => {
                log::info!("[C <- S] on_stream_is_recorded...");
                // 调用处理流已录制消息的方法
                self.on_stream_is_recorded(stream_id)?;
            }
            // 处理音频数据消息
            RtmpMessageData::AudioData { data } => {
                // 调用处理音频数据的方法
                self.common.on_audio_data(data, timestamp).await?
            }
            // 处理视频数据消息
            RtmpMessageData::VideoData { data } => {
                // 调用处理视频数据的方法
                self.common.on_video_data(data, timestamp).await?
            }
            // 处理 AMF 数据消息
            RtmpMessageData::AmfData { raw_data } => {
                // 调用处理元数据的方法
                self.common.on_metadata(raw_data, timestamp).await?;
            }
            // 其他未处理的消息类型
            _ => {}
        }
        Ok(())
    }

    // 异步处理 AMF0 命令消息
    pub async fn on_amf0_command_message(
        &mut self,
        command_name: &Amf0ValueType,
        transaction_id: &Amf0ValueType,
        command_object: &Amf0ValueType,
        others: &mut Vec<Amf0ValueType>,
    ) -> Result<(), SessionError> {
        log::info!("[C <- S] on_amf0_command_message...");
        // 获取命令名称
        let empty_cmd_name = &String::new();
        let cmd_name = match command_name {
            Amf0ValueType::UTF8String(str) => str,
            _ => empty_cmd_name,
        };

        // 获取事务 ID
        let transaction_id = match transaction_id {
            Amf0ValueType::Number(number) => *number as u8,
            _ => 0,
        };

        let empty_cmd_obj: IndexMap<String, Amf0ValueType> = IndexMap::new();
        let _ = match command_object {
            Amf0ValueType::Object(obj) => obj,
            // Amf0ValueType::Null =>
            _ => &empty_cmd_obj,
        };

        // 根据命令名称和事务 ID 进行处理
        match cmd_name.as_str() {
            "_result" => match transaction_id {
                define::TRANSACTION_ID_CONNECT => {
                    log::info!("[C <- S] on_result_connect...");
                    self.on_result_connect().await?;
                }
                define::TRANSACTION_ID_CREATE_STREAM => {
                    log::info!("[C <- S] on_result_create_stream...");
                    self.on_result_create_stream()?;
                }
                _ => {}
            },
            "_error" => {
                self.on_error()?;
            }
            "onStatus" => {
                match others.remove(0) {
                    Amf0ValueType::Object(obj) => self.on_status(&obj).await?,
                    _ => {
                        return Err(SessionError {
                            value: SessionErrorValue::Amf0ValueCountNotCorrectError,
                        })
                    }
                };
            }

            _ => {}
        }

        Ok(())
    }

    // 异步发送连接请求
    pub async fn send_connect(&mut self, transaction_id: &f64) -> Result<(), SessionError> {
        // 先发送快的大小的消息
        self.send_set_chunk_size().await?;

        // 创建网络连接实例
        let mut netconnection = NetConnection::new(Arc::clone(&self.io));
        // 创建连接属性实例
        let mut properties = ConnectProperties::new_none();

        // 构建 RTMP 的 URL
        let url = format!(
            "rtmp://{domain_name}/{app_name}",
            domain_name = self.raw_domain_name,
            app_name = self.app_name
        );
        // 设置应用名称
        properties.app = Some(self.app_name.clone());

        // 根据客户端类型设置不同的连接属性
        match self.client_type {
            ClientSessionType::Pull => {
                properties.flash_ver = Some("LNX 9,0,124,2".to_string());
                properties.tc_url = Some(url.clone());
                properties.fpad = Some(false);
                properties.capabilities = Some(15_f64);
                properties.audio_codecs = Some(4071_f64);
                properties.video_codecs = Some(252_f64);
                properties.video_function = Some(1_f64);
            }
            ClientSessionType::Push => {
                properties.pub_type = Some("nonprivate".to_string());
                properties.flash_ver = Some("FMLE/3.0 (compatible; xiu)".to_string());
                properties.fpad = Some(false);
                properties.tc_url = Some(url.clone());
            }
        }

        // 发送连接请求
        netconnection
            .write_connect(transaction_id, &properties)
            .await?;

        Ok(())
    }

    // 异步发送创建流请求
    pub async fn send_create_stream(&mut self, transaction_id: &f64) -> Result<(), SessionError> {
        let mut netconnection = NetConnection::new(Arc::clone(&self.io));
        netconnection.write_create_stream(transaction_id).await?;

        Ok(())
    }

    // 异步发送删除流请求
    pub async fn send_delete_stream(
        &mut self,
        transaction_id: &f64,
        stream_id: &f64,
    ) -> Result<(), SessionError> {
        let mut netstream = NetStreamWriter::new(Arc::clone(&self.io));
        netstream
            .write_delete_stream(transaction_id, stream_id)
            .await?;

        Ok(())
    }

    // 异步发送发布请求
    pub async fn send_publish(
        &mut self,
        transaction_id: &f64,
        stream_name: &String,
        stream_type: &String,
    ) -> Result<(), SessionError> {
        let mut netstream = NetStreamWriter::new(Arc::clone(&self.io));
        netstream
            .write_publish(transaction_id, stream_name, stream_type)
            .await?;

        Ok(())
    }

    // 异步发送播放请求
    pub async fn send_play(
        &mut self,
        transaction_id: &f64,
        stream_name: &String,
        start: &f64,
        duration: &f64,
        reset: &bool,
    ) -> Result<(), SessionError> {
        let mut netstream = NetStreamWriter::new(Arc::clone(&self.io));
        netstream
            .write_play(transaction_id, stream_name, start, duration, reset)
            .await?;

        let mut netconnection = NetConnection::new(Arc::clone(&self.io));
        netconnection
            .write_get_stream_length(transaction_id, stream_name)
            .await?;

        self.send_set_buffer_length(1, 1300).await?;

        Ok(())
    }

    // 异步发送设置块大小消息
    pub async fn send_set_chunk_size(&mut self) -> Result<(), SessionError> {
        let mut controlmessage =
            ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
        controlmessage.write_set_chunk_size(CHUNK_SIZE).await?;
        Ok(())
    }

    // 异步发送窗口确认大小消息
    pub async fn send_window_acknowledgement_size(
        &mut self,
        window_size: u32,
    ) -> Result<(), SessionError> {
        let mut controlmessage =
            ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
        controlmessage
            .write_window_acknowledgement_size(window_size)
            .await?;
        Ok(())
    }

    // 异步发送设置缓冲区长度消息
    pub async fn send_set_buffer_length(
        &mut self,
        stream_id: u32,
        ms: u32,
    ) -> Result<(), SessionError> {
        let mut eventmessages = EventMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
        eventmessages.write_set_buffer_length(stream_id, ms).await?;

        Ok(())
    }

    // 异步处理连接成功的结果
    pub async fn on_result_connect(&mut self) -> Result<(), SessionError> {
        let mut controlmessage =
            ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
        controlmessage.write_acknowledgement(3107).await?;

        let mut netstream = NetStreamWriter::new(Arc::clone(&self.io));
        netstream
            .write_release_stream(&(define::TRANSACTION_ID_CONNECT as f64), &self.stream_name)
            .await?;
        netstream
            .write_fcpublish(&(define::TRANSACTION_ID_CONNECT as f64), &self.stream_name)
            .await?;

        self.state = ClientSessionState::CreateStream;

        Ok(())
    }

    // 处理创建流成功的结果
    pub fn on_result_create_stream(&mut self) -> Result<(), SessionError> {
        match self.client_type {
            ClientSessionType::Pull => {
                self.state = ClientSessionState::Play;
            }
            ClientSessionType::Push => {
                self.state = ClientSessionState::PublishingContent;
            }
        }
        Ok(())
    }

    // 处理设置块大小消息
    pub fn on_set_chunk_size(&mut self, chunk_size: &mut u32) -> Result<(), SessionError> {
        self.unpacketizer
            .update_max_chunk_size(*chunk_size as usize);
        Ok(())
    }

    // 处理流已录制消息
    pub fn on_stream_is_recorded(&mut self, stream_id: &mut u32) -> Result<(), SessionError> {
        log::trace!("stream is recorded stream_id is {}", stream_id);
        Ok(())
    }

    // 处理流开始消息
    pub fn on_stream_begin(&mut self, stream_id: &mut u32) -> Result<(), SessionError> {
        log::trace!("stream is begin stream_id is {}", stream_id);
        Ok(())
    }

    // 异步处理设置对等带宽消息
    pub async fn on_set_peer_bandwidth(&mut self) -> Result<(), SessionError> {
        self.send_window_acknowledgement_size(5000000).await?;

        Ok(())
    }

    // 处理错误消息
    pub fn on_error(&mut self) -> Result<(), SessionError> {
        Ok(())
    }

    // 异步处理状态消息
    pub async fn on_status(
        &mut self,
        obj: &IndexMap<String, Amf0ValueType>,
    ) -> Result<(), SessionError> {
        if let Some(Amf0ValueType::UTF8String(code_info)) = obj.get("code") {
            match &code_info[..] {
                // 处理开始发布状态
                "NetStream.Publish.Start" => {
                    // 状态切换到开始发布状态
                    self.state = ClientSessionState::StartPublish;
                    // 根据子应用和子流名称订阅流
                    if let (Some(app_name), Some(stream_name)) =
                        (&self.sub_app_name, &self.sub_stream_name)
                    {
                        self.common
                            .subscribe_from_stream_hub(app_name.clone(), stream_name.clone())
                            .await?;
                    } else {
                        self.common
                            .subscribe_from_stream_hub(
                                self.app_name.clone(),
                                self.stream_name.clone(),
                            )
                            .await?;
                    }
                }
                // 处理发布重置状态
                "NetStream.Publish.Reset" => {}
                // 处理开始播放状态
                "NetStream.Play.Start" => {
                    // 从远程 RTMP 服务器拉取流并发布到本地会话
                    self.common
                        .publish_to_stream_hub(
                            self.app_name.clone(),
                            self.stream_name.clone(),
                            self.gop_num,
                        )
                        .await?
                }
                _ => {}
            }
        }
        log::trace!("{}", obj.len());
        Ok(())
    }

    // 订阅指定的应用和流
    pub fn subscribe(&mut self, app_name: String, stream_name: String) {
        self.sub_app_name = Some(app_name);
        self.sub_stream_name = Some(stream_name);
    }
}
